package com.burges.net.tableAPIAndSQL.table

import org.apache.flink.streaming.api.scala.StreamExecutionEnvironment
import org.apache.flink.table.api.TableEnvironment
import org.apache.flink.table.descriptors.{FileSystem, Kafka}

/**
  * 创建人    BurgessLee 
  * 创建时间   2020/2/13 
  * 描述     TableConnector
  */
object TableConnector {

	def main(args: Array[String]): Unit = {
		val environment = StreamExecutionEnvironment.getExecutionEnvironment
		val tableEnvironment = TableEnvironment.getTableEnvironment(environment)
		//文件系统FileSystem
		tableEnvironment.connect(
			new FileSystem().path("file:///path/filename")//可以是文件夹或者是文件
		)

		// kafka connector
		tableEnvironment.connect(
			new Kafka()
        			.version("0.11") //指定版本，0.8 0.9 0.10 0.11
        			.topic("myTopic") //指定Table对应的KafkaTopic
					// 通过Property指定KafkaConnector需要的配置信息
        			.property("zookeeper.connect", "localhost:2181")
        			.property("boostrap.servers", "localhost:9092")
        			.property("group.id", "KafkaGroup")
        			// 从Kafka中读取数据：指定Offset的启动模式（可选）
        			.startFromEarliest() //从最早的Offset开始消费
        			.startFromLatest()   //从最新的Offset开始消费
//        			.startFromGroupOffsets(...) //从指定的offset开始消费
        			.sinkPartitionerFixed() //每个Flink分区最多被分配到一个Kakafa分区上
        			.sinkPartitionerRoundRobin() //Flink中分区随机映射到Kafka分区上
//        			.sinkPartitionerCustom(CustomPartitioner.class) //自定义KafkaPartitioner
		)
	}


}
